package com.ehotting.mqbd.core.manager;

import com.ehotting.eaf.redis.service.RedisService;
import com.ehotting.mqbd.api.MQDBConstants;
import com.ehotting.mqbd.api.bean.ConsumerNodeBean;
import com.ehotting.mqbd.api.enums.ConsumerNodeStateEnum;
import com.ehotting.mqbd.core.service.ConsumerNodeService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.stream.Collectors;

@Component
public class ConsumerNodeManager {

    private static Logger logger = LoggerFactory.getLogger(ConsumerNodeManager.class);

    @Autowired
    private ConsumerNodeService consumerNodeService;

    private volatile boolean isRun = true;

    private Thread thread = null;

    public void save(ConsumerNodeBean consumerNode){
        consumerNodeService.save(consumerNode);
    }

    public void delete(String ip){
        consumerNodeService.delete(ip);
    }

    private void addCount(String ip){
        consumerNodeService.addCount(ip);
    }

    public boolean isExits(String ip){
        ConsumerNodeBean consumerNode = getLive(ip);
        return consumerNode!=null;
    }

    public ConsumerNodeBean getLive(String ip){
        List<ConsumerNodeBean>  list = consumerNodeService.listAll();
        return list.stream()
                .filter(t->{return t.getIp().equals(ip) && t.getState() == ConsumerNodeStateEnum.YES;})
                .sorted(Comparator.comparing(ConsumerNodeBean::getCount))
                .findFirst()
                .orElse(null);
    }



    public List<ConsumerNodeBean> getLiveList(){
        List<ConsumerNodeBean>  list = consumerNodeService.listAll();
        return  list.stream()
                .filter(t->{return  t.getState() == ConsumerNodeStateEnum.YES;})
                .collect(Collectors.toList());
    }




    public void clearOffLineConsumer(){
        logger.info("Start clearOffLineConsumer.");
        List<ConsumerNodeBean> consumerNodeList =  getLiveList();
        if(consumerNodeList == null){
            logger.error("consumerNodeList is null,sleep 10s,continue.");
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return;
        }
        for(ConsumerNodeBean consumerNode : consumerNodeList){
            if(consumerNode == null){
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                continue;
            }
            if(consumerNode.getDate() == null){
                delete(consumerNode.getIp());
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                logger.error("consumerNode.getDate() is null,delete ip[{}],sleep 10s,continue.",consumerNode.getIp());
                continue;
            }

            if(Math.abs(consumerNode.getDate().getTime()/1000-(new Date()).getTime()/1000)>=60){
                delete(consumerNode.getIp());
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                logger.error("consumerNode.getDate() is out times 60s,delete ip[{}],sleep 10s,continue.",consumerNode.getIp());
            }
        }
        logger.info("Finish clearOffLineConsumer.");
    }


}
